package com.gitee.dbswitch.data.service;

import com.gitee.dbswitch.common.entity.CloseableDataSource;
import com.gitee.dbswitch.common.entity.LoggingSupplier;
import com.gitee.dbswitch.common.entity.MdcKeyValue;
import com.gitee.dbswitch.core.robot.RobotAPI;
import com.gitee.dbswitch.data.config.DbswtichPropertiesConfiguration;
import com.gitee.dbswitch.data.domain.APIHttpParam;
import com.gitee.dbswitch.data.domain.APITaskResult;
import com.gitee.dbswitch.data.handler.APIReaderTaskThread;
import com.gitee.dbswitch.schema.TableDescription;
import com.gitee.dbswitch.service.DefaultMetadataService;
import com.gitee.dbswitch.service.MetadataService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.core.task.AsyncTaskExecutor;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.function.Supplier;
import java.util.stream.Collectors;

/**
 * 目标数据库表并发写入控制
 *
 * @author natural
 */
@Slf4j
public class DefaultAPIReaderRobot extends RobotAPI<APITaskResult> {

    private CountDownLatch apiCountDownLatch;
    private final MdcKeyValue mdcKeyValue;
    private AsyncTaskExecutor threadExecutor;
    private List<Supplier> apiHttpThreads;
    private List<CompletableFuture> futures;
    private DbswtichPropertiesConfiguration configuration;
    private CloseableDataSource targetDataSource;

    public DefaultAPIReaderRobot(MdcKeyValue mdcKeyValue,
                                 DbswtichPropertiesConfiguration configuration,
                                 CloseableDataSource targetDataSource) {
        this.mdcKeyValue = mdcKeyValue;
        this.configuration = configuration;
        this.targetDataSource = targetDataSource;
    }

    @Override
    public void interrupt() {
        Optional.ofNullable(futures).orElseGet(ArrayList::new).forEach(f -> f.cancel(true));
    }

    @Override
    public void init(AsyncTaskExecutor threadExecutor) {
        this.threadExecutor = threadExecutor;
        this.apiHttpThreads = new ArrayList<>();
        this.apiCountDownLatch = new CountDownLatch(1);
        MetadataService tdsService = new DefaultMetadataService(targetDataSource);
        Set<String> targetExistTables = tdsService.queryTableList(configuration.getTarget().getTargetSchema())
                .stream().map(TableDescription::getTableName).collect(Collectors.toSet());
        APIHttpParam param = APIHttpParam.builder()
                .memChannel(this.getChannel())
                .apiURL(configuration.getApi().getApiUrl())
                .requestMethod(configuration.getApi().getRequestMethod())
                .synchronizeType(configuration.getApi().getSynchronizeType())
                .apiParams(configuration.getApi().getApiParams())
                .apiBody(configuration.getApi().getApiBody())
                .apiHeaders(configuration.getApi().getApiHeaders())
                .apiJsonRoot(configuration.getApi().getApiJsonRoot())
                .apiJsonPosition(configuration.getApi().getApiJsonPosition())
                .targetDataSource(targetDataSource)
                .targetExistTables(targetExistTables)
                .configuration(configuration)
                .apiCountDownLatch(apiCountDownLatch)
                .build();
        if (Objects.nonNull(mdcKeyValue)) {
            apiHttpThreads.add(new LoggingSupplier(new APIReaderTaskThread(param), mdcKeyValue));
        } else {
            apiHttpThreads.add(new APIReaderTaskThread(param));
        }

    }

    @Override
    public void startHttp() {
        futures = new ArrayList<>(apiHttpThreads.size());
        apiHttpThreads.forEach(
                task ->
                        futures.add(CompletableFuture.supplyAsync(task, threadExecutor))
        );
    }

    @Override
    public Optional<APITaskResult> getWorkResult() {

        return futures.stream().map(CompletableFuture::join)
                .filter(Objects::nonNull)
                .map(f -> (APITaskResult) f)
                .peek(APITaskResult::padding)
                .reduce(
                        (r1, r2) -> {
                            String resultHttp = r1.getAPIResponse();
                            return APITaskResult.builder()
                                    .APIResponse(resultHttp)
                                    .build();
                        }
                );
    }


    @Override
    public long getRemainingCount() {
        return this.apiCountDownLatch.getCount();
    }
}
